package com.ds.lens.flink.monitor.common;

import com.ds.lens.data.common.document.dto.MonitorEvent;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;

public class MonitorEventSerializeSchema extends CommonSerializationSchema<MonitorEvent> {
    @Override
    public byte[] serializeKey(MonitorEvent event) {
        if (null != event.getMonitorKey() && null != event.getMonitorKey().getMonitorId()
                && null != event.getTriggered() && StringUtils.isNotEmpty(event.getTriggered().getGroupValue())) {
            Long monitorId = event.getMonitorKey().getMonitorId();
            String groupValue = event.getTriggered().getGroupValue();
            StringBuilder builder = new StringBuilder(monitorId.toString()).append(groupValue);
            return builder.toString().getBytes(StandardCharsets.UTF_8);
        }
        return String.valueOf(System.currentTimeMillis()).getBytes(StandardCharsets.UTF_8);
    }
}
